/**
 * JDBM LICENSE v1.00
 *
 * Redistribution and use of this software and associated documentation
 * ("Software"), with or without modification, are permitted provided
 * that the following conditions are met:
 *
 * 1. Redistributions of source code must retain copyright
 *    statements and notices.  Redistributions must also contain a
 *    copy of this document.
 *
 * 2. Redistributions in binary form must reproduce the
 *    above copyright notice, this list of conditions and the
 *    following disclaimer in the documentation and/or other
 *    materials provided with the distribution.
 *
 * 3. The name "JDBM" must not be used to endorse or promote
 *    products derived from this Software without prior written
 *    permission of Cees de Groot.  For written permission,
 *    please contact cg@cdegroot.com.
 *
 * 4. Products derived from this Software may not be called "JDBM"
 *    nor may "JDBM" appear in their names without prior written
 *    permission of Cees de Groot.
 *
 * 5. Due credit should be given to the JDBM Project
 *    (http://jdbm.sourceforge.net/).
 *
 * THIS SOFTWARE IS PROVIDED BY THE JDBM PROJECT AND CONTRIBUTORS
 * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
 * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
 * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
 * CEES DE GROOT OR ANY CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * Copyright 2000 (C) Cees de Groot. All Rights Reserved.
 * Contributions are Copyright (C) 2000 by their associated contributors.
 *
 * $Id: TransactionManager.java,v 1.7 2005/06/25 23:12:32 doomdark Exp $
 */

package com.sunchangming.stmdb.recman;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.TreeSet;

import com.google.protobuf.CodedInputStream;
import com.google.protobuf.CodedOutputStream;

/**
 * TODO：现在是每个RecordFile对应一个TransactionManager，得改掉。
 * 
 * @author cm
 * 
 */
public final class TransactionManager {
	private final RecordFile owner;

	// streams for transaction log.
	private FileOutputStream fos;
	private CodedOutputStream oos;

	/**
	 * By default, we keep 10 transactions in the log file before synchronizing
	 * it with the main database file.
	 */
	static final int DEFAULT_TXNS_IN_LOG = 10;

	/**
	 * Maximum number of transactions before the log file is synchronized with
	 * the main database file.
	 */
	private final int _maxTxns = DEFAULT_TXNS_IN_LOG;

	/**
	 * In-core copy of transactions. We could read everything back from the log
	 * file, but the RecordFile needs to keep the dirty blocks in core anyway,
	 * so we might as well point to them and spare us a lot of hassle.
	 */
	@SuppressWarnings("unchecked")
	private final ArrayList<BlockIo>[] txns = new ArrayList[DEFAULT_TXNS_IN_LOG];
	private int curTxn = -1;

	/** Extension of a log file. */
	static final String extension = ".lg";

	/**
	 * Instantiates a transaction manager instance. If recovery needs to be
	 * performed, it is done.
	 * 
	 * @param owner
	 *            the RecordFile instance that owns this transaction mgr.
	 */
	TransactionManager(RecordFile owner) throws IOException {
		this.owner = owner;
		recover();
		open();
	}

	/** Builds logfile name */
	private String makeLogName() {
		return owner.getFileName() + extension;
	}

	/**
	 * 写数据文件。
	 * 
	 * @throws IOException
	 */
	private void synchronizeLogFromMemory() throws IOException {
		//因为日志文件已经写完了，先把日志文件关闭。
		close();

		final TreeSet<BlockIo> blockList = new TreeSet<BlockIo>(new BlockIoComparator());

		for (int i = 0; i < _maxTxns; i++) {
			if (txns[i] == null)
				continue;
			// Add each block to the blockList, replacing the old copy of this
			// block if necessary, thus avoiding writing the same block twice
			for (final Iterator<BlockIo> k = txns[i].iterator(); k.hasNext();) {
				final BlockIo block = k.next();
				if (blockList.contains(block)) {
					block.decrementTransactionCount();
				} else {
					blockList.add(block);
				}
			}

			txns[i] = null;
		}
		// 把blocks按blockid顺序写入到数据文件中
		synchronizeBlocks(blockList.iterator(), true);
		//同步数据文件
		owner.sync();
		//开一个新的日志文件
		open();
	}

	/** Opens the log file */
	private void open() throws IOException {
		final String filename = makeLogName();
		final java.nio.file.Path path = java.nio.file.Paths.get(filename);
		fos = new FileOutputStream(path.toFile());
		oos = CodedOutputStream.newInstance(fos);
		oos.writeFixed32NoTag(Magic.LOGFILE_HEADER);
		oos.flush();
		curTxn = -1;
	}

	private void recover() throws IOException {
		final String logName = makeLogName();
		final java.nio.file.Path path = java.nio.file.Paths.get(logName);

		if (!java.nio.file.Files.exists(path))
			return;

		if (java.nio.file.Files.size(path) == 0) {
			java.nio.file.Files.delete(path);
			return;
		}

		final FileInputStream fis = new FileInputStream(path.toFile());
		final CodedInputStream ois = CodedInputStream.newInstance(fis);

		try {
			if (ois.readFixed32() != Magic.LOGFILE_HEADER)
				throw new Error("Bad magic on log file");
		} catch (final IOException e) {
			java.nio.file.Files.delete(path);
			return;
		}

		while (!ois.isAtEnd()) {
			ArrayList<BlockIo> blocks = null;
			try {
				final int length = ois.readInt32();
				blocks = new ArrayList<BlockIo>(length);
				for (int i = 0; i != length; ++i) {
					blocks.add(BlockIo.read(ois));
				}
			} catch (final IOException e) {
				break;
			}

			synchronizeBlocks(blocks.iterator(), false);
		}

		owner.sync();
		java.nio.file.Files.delete(path);
	}

	private void synchronizeBlocks(Iterator<BlockIo> blockIterator, boolean fromCore) throws IOException {
		// write block vector elements to the data file.
		while (blockIterator.hasNext()) {
			final BlockIo cur = blockIterator.next();
			owner.synch(cur);
			if (fromCore) {
				cur.decrementTransactionCount();
				if (!cur.isInTransaction()) {
					owner.releaseFromTransaction(cur);
				}
			}
		}
	}

	/** Set clean flag on the blocks. */
	private void setClean(ArrayList<BlockIo> blocks) throws IOException {
		for (final Iterator<BlockIo> k = blocks.iterator(); k.hasNext();) {
			final BlockIo cur = k.next();
			cur.setClean();
		}
	}

	/** Discards the indicated blocks and notify the owner. */
	private void discardBlocks(ArrayList<BlockIo> blocks) throws IOException {
		for (final Iterator<BlockIo> k = blocks.iterator(); k.hasNext();) {
			final BlockIo cur = k.next();
			cur.decrementTransactionCount();
			if (!cur.isInTransaction()) {
				owner.releaseFromTransaction(cur);
			}
		}
	}

	/**
	 * 开始一个新事务。
	 * 
	 * @throws IOException
	 */
	void start() throws IOException {
		curTxn++;
		if (curTxn == _maxTxns) {
			synchronizeLogFromMemory();
			curTxn = 0;
		}
		txns[curTxn] = new ArrayList<BlockIo>();
	}

	/**
	 * Indicates the block is part of the transaction.
	 */
	void add(BlockIo block) throws IOException {
		block.incrementTransactionCount();
		txns[curTxn].add(block);
	}

	/**
	 * 把当前事务(curTxn)写入到日志文件中
	 * 
	 * @throws IOException
	 */
	void commit() throws IOException {
		final ArrayList<BlockIo> bios = txns[curTxn];
		oos.writeInt32NoTag(bios.size());
		for (final BlockIo b : bios) {
			b.write(oos);
		}
		sync();
		// set clean flag to indicate blocks have been written to log
		setClean(txns[curTxn]);
	}

	/** Flushes and syncs */
	private void sync() throws IOException {
		oos.flush();
		fos.flush();
		fos.getFD().sync();
	}

	/**
	 * Shutdowns the transaction manager. Resynchronizes outstanding logs.
	 */
	void shutdown() throws IOException {
		synchronizeLogFromMemory();
		close();
	}

	/**
	 * Closes open files.
	 */
	private void close() throws IOException {
		sync();
		oos.flush();
		fos.close();
		oos = null;
		fos = null;
	}

	/**
	 * Force closing the file without synchronizing pending transaction data.
	 * Used for testing purposes only.
	 */
	void forceClose() throws IOException {
		oos.flush();
		fos.close();
		oos = null;
		fos = null;
	}

	/**
	 * Use the disk-based transaction log to synchronize the data file.
	 * Outstanding memory logs are discarded because they are believed to be
	 * inconsistent.
	 */
	void synchronizeLogFromDisk() throws IOException {
		close();

		for (int i = 0; i < _maxTxns; i++) {
			if (txns[i] == null)
				continue;
			discardBlocks(txns[i]);
			txns[i] = null;
		}

		recover();
		open();
	}

	/**
	 * 把BlockIo按照blockid排序
	 * 
	 * @author Changming
	 * 
	 */
	public static class BlockIoComparator implements Comparator<BlockIo> {

		@Override
		public int compare(BlockIo block1, BlockIo block2) {
			int result = 0;
			if (block1.getBlockId() == block2.getBlockId()) {
				result = 0;
			} else if (block1.getBlockId() < block2.getBlockId()) {
				result = -1;
			} else {
				result = 1;
			}
			return result;
		}
	}

}
